package com.lzcu.maoyan.mapreduce.mapreduce;

import com.lzcu.maoyan.mapreduce.mapreduce.mapper.MovieMapper;
import com.lzcu.maoyan.mapreduce.mapreduce.reduce.MovieReduce;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 电影信息入库清洗
 *
 * @author WangLiang
 * @date `2024/3/25`
 */
public class MovieAnalysisAndStorage {

    public static void main(String[] args) throws Exception {
        movieDataCleaning();
    }

    public static void movieDataCleaning() throws Exception {
        System.setProperty("HADOOP_USER_NAME", "hadoop");
        // 有时设置无效
        System.setProperty("user.name", "hadoop");
        String args1 = "hdfs://192.168.2.109:9000/maoyan_movie/realtime_movie/part-m-00000";
        String args2 = "hdfs://192.168.2.109:9000/maoyan_movie/output";
        //1:获取job信息
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration, "maoyan_movie-data-cleaning");
        //2：获取jar包信息
        job.setJarByClass(MovieAnalysisAndStorage.class);
        //3：关联自定义的mapper和reducer
        job.setMapperClass(MovieMapper.class);
        job.setReducerClass(MovieReduce.class);
        //4：设置map输出类型数据
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        //5：设置最终输出数据类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        //以下的的四行代码设立的原因是为了解决你的输出目录已经存在文件的情况，如果你的输出目录已经存在文件夹，那么你是无法进行输出的
        //所以我们就对进行检测，如果存在就把他删了
        // 取第1个表示输出目录参数（第0个参数是输入目录）
        Path path = new Path(args2);
        // 根据path找到这个文件
        FileSystem fileSystem = path.getFileSystem(configuration);
        if (fileSystem.exists(path)) {
            // true的意思是，就算output有东西，也一带删除
            fileSystem.delete(path, true);
        }
        //6：设置数据输入和输出文件路径
        FileInputFormat.setInputPaths(job, new Path(args1));
        FileOutputFormat.setOutputPath(job, new Path(args2));
        job.waitForCompletion(true);
    }
}